package com.jwater.core.serialize;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingDeque;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.jwater.core.common.S;

import jwater.io.protostuff.LinkedBuffer;
import jwater.io.protostuff.ProtostuffIOUtil;
import jwater.io.protostuff.Schema;
import jwater.io.protostuff.runtime.RuntimeSchema;
import jwater.org.apache.mina.core.buffer.IoBuffer;
/**
 * @description:
 * @author: jnner
 * @createDate: 2021/2/5
 * @version: 1.0
 */
public class ProtoStuffSerialize extends AbstractSerialize {
	protected static Logger logger = LoggerFactory.getLogger(ProtoStuffSerialize.class);

	/**
	 * 避免每次序列化都重新申请Buffer空间
	 */
	private static LinkedBlockingDeque<LinkedBuffer> bufferCache = new LinkedBlockingDeque<>();
	/**
	 * 缓存Schema
	 */
	private static Map<Class<?>, Schema<?>> schemaCache = new ConcurrentHashMap<>();
	/**
	 * 缓存clazz
	 */
	private static Map<String, Class<?>> classNameCache = new ConcurrentHashMap<>();

	/**
	 * 序列化
	 * @param obj
	 * @param <T>
	 * @return
	 * @throws SerializeException
	 */
	@SuppressWarnings("unchecked")
	@Override
	public <T> byte[] serialize(T obj) throws SerializeException {
		LinkedBuffer buffer = getBuffer();
		try {
			Class<T> clazz = (Class<T>) obj.getClass();
			Schema<T> schema = getSchema(clazz);
			byte[] data;
			try {
				data = ProtostuffIOUtil.toByteArray(obj, schema, buffer);
			} finally {
				buffer.clear();
			}
			return data;
		} catch (Exception e) {
			throw new SerializeException("ProtoStuffSerialize error", e);
		} finally {
			bufferCache.addLast(buffer);
		}
	}

	/**
	 * 反序列化方法，将字节数组反序列化成指定Class类型
	 *
	 * @param data
	 * @param clazz
	 * @param <T>
	 * @return
	 */
	@Override
	public <T> T deserialize(byte[] data, Class<T> clazz) throws SerializeException {
		try {
			Schema<T> schema = getSchema(clazz);
			T obj = schema.newMessage();
			ProtostuffIOUtil.mergeFrom(data, obj, schema);
			return obj;
		} catch (Exception e) {
			throw new SerializeException("ProtoStuffDeserialize error", e);
		}
	}

	@SuppressWarnings("unchecked")
	private <T> Schema<T> getSchema(Class<T> clazz) {
		Schema<T> schema = (Schema<T>) schemaCache.get(clazz);
		if (Objects.isNull(schema)) {
			// 这个schema通过RuntimeSchema进行懒创建并缓存
			// 所以可以一直调用RuntimeSchema.getSchema(),这个方法是线程安全的
			schema = RuntimeSchema.getSchema(clazz);
			if (Objects.nonNull(schema)) {
				schemaCache.put(clazz, schema);
			}
		}
		return schema;
	}

	private LinkedBuffer getBuffer() {
		if (bufferCache.size() < 20) {
			bufferCache.addLast(LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE));
		}
		LinkedBuffer buffer = null;
		try {
			buffer = bufferCache.takeFirst();
		} catch (InterruptedException e) {
			logger.error(e.toString(), e);
		}
		return buffer;
	}

	@Override
	public byte[] serializeArray(Object... objs) throws SerializeException {
		try {
			if (objs == null || objs.length == 0) {
				return new byte[0];
			}
			List<byte[]> classNames = new ArrayList<byte[]>();
			List<byte[]> datas = new ArrayList<byte[]>();
			int len = 0;
			for (Object obj : objs) {
				byte[] className = obj.getClass().getCanonicalName().getBytes(S.charset);
				classNames.add(className);
				len += className.length;
				byte[] data = serialize(obj);
				datas.add(data);
				len += data.length;
			}
			IoBuffer buffer = IoBuffer.allocate(len + (classNames.size() + datas.size()) * 4);
			int i = 0;
			for (byte[] data : datas) {
				byte[] classNameBytes = classNames.get(i);
				buffer.putInt(classNameBytes.length);
				buffer.put(classNameBytes);
				buffer.putInt(data.length);
				buffer.put(data);
				i++;
			}
			return buffer.array();
		} catch (Exception e) {
			throw new SerializeException("ProtoStuff SerializeArray error " + e.getMessage(), e);
		}

	}

	@Override
	public Object[] deserializeArray(byte[] data) throws SerializeException {
		try {
			if (data == null || data.length == 0) {
				return new Object[0];
			}
			List<Object> list = new ArrayList<Object>();
			IoBuffer buffer = IoBuffer.wrap(data);
			while (buffer.hasRemaining()) {
				int len = buffer.getInt();
				byte[] bytes = new byte[len];
				buffer.get(bytes);
				String className = new String(bytes, 0, bytes.length, S.charset);
				Class<?> clazz = classNameCache.get(className);
				if (clazz == null) {
					clazz = Class.forName(className);
					classNameCache.put(className, clazz);
				}
				len = buffer.getInt();
				bytes = new byte[len];
				buffer.get(bytes);
				list.add(deserialize(bytes, clazz));
			}
			return list.toArray();
		} catch (Exception e) {
			throw new SerializeException("ProtoStuff DeserializeArray error " + e.getMessage(), e);
		}
	}
}
